#pragma once

#include <Poco/DateTimeFormatter.h>
#include <Poco/Net/NetException.h>
#include <Poco/Net/ServerSocket.h>
#include <Poco/Net/SocketAddress.h>
#include <Poco/Net/StreamSocket.h>
#include <Poco/Net/TCPServer.h>
#include <Poco/Net/TCPServerParams.h>
#include <Poco/Timestamp.h>
#include <Poco/Util/Application.h>
#include <Poco/Util/ServerApplication.h>
#if defined(POCO_HAVE_FD_EPOLL)
#include <sys/epoll.h>
#else
#include <sys/poll.h>
#endif
#include <unistd.h>

#include "Common/Logger.h"
#include "FServer/MySQL/MySQLRequest.h"
#include "MServer/MySQL/MySQLPacket.h"


namespace keeper_server {

struct PollResult {
  size_t responses_count{0};
  bool has_requests{false};
  bool error{false};
};

struct SocketInterruptablePollWrapper {
  int sockfd;

#if defined(POCO_HAVE_FD_EPOLL)
  int epollfd;
  epoll_event socket_event{};
  epoll_event pipe_event{};
#endif

  explicit SocketInterruptablePollWrapper(
      const Poco::Net::StreamSocket& poco_socket_)
      : sockfd(poco_socket_.impl()->sockfd()) {
#if defined(POCO_HAVE_FD_EPOLL)
    epollfd = epoll_create(2);
    if (epollfd < 0)
      throwFromErrno("Cannot epoll_create", ErrorCodes::SYSTEM_ERROR);

    socket_event.events = EPOLLIN | EPOLLERR | EPOLLPRI;
    socket_event.data.fd = sockfd;
    if (epoll_ctl(epollfd, EPOLL_CTL_ADD, sockfd, &socket_event) < 0) {
      ::close(epollfd);
      throwFromErrno("Cannot insert socket into epoll queue",
                     ErrorCodes::SYSTEM_ERROR);
    }
#endif
  }

  PollResult poll(Poco::Timespan remaining_time) {
    bool socket_ready = false;
    bool fd_ready = false;

    int rc = 0;
    if (!fd_ready) {
#if defined(POCO_HAVE_FD_EPOLL)
      epoll_event evout[2];
      evout[0].data.fd = evout[1].data.fd = -1;
      do {
        Poco::Timestamp start;
        rc = epoll_wait(epollfd, evout, 2, remaining_time.totalMilliseconds());
        if (rc < 0 && errno == EINTR) {
          Poco::Timestamp end;
          Poco::Timespan waited = end - start;
          if (waited < remaining_time)
            remaining_time -= waited;
          else
            remaining_time = 0;
        }
      } while (rc < 0 && errno == EINTR);

      for (int i = 0; i < rc; ++i) {
        if (evout[i].data.fd == sockfd) socket_ready = true;
      }
#else
      pollfd poll_buf[2];
      poll_buf[0].fd = sockfd;
      poll_buf[0].events = POLLIN;

      do {
        Poco::Timestamp start;
        rc = ::poll(poll_buf, 2, remaining_time.totalMilliseconds());
        if (rc < 0 && errno == POCO_EINTR) {
          Poco::Timestamp end;
          Poco::Timespan waited = end - start;
          if (waited < remaining_time)
            remaining_time -= waited;
          else
            remaining_time = 0;
        }
      } while (rc < 0 && errno == POCO_EINTR);

      if (rc >= 1 && poll_buf[0].revents & POLLIN) socket_ready = true;
#endif
    }

    PollResult result{};
    result.has_requests = socket_ready;

    if (rc < 0) result.error = true;

    return result;
  }

#if defined(POCO_HAVE_FD_EPOLL)
  ~SocketInterruptablePollWrapper() { ::close(epollfd); }
#endif
};

using SocketInterruptablePollWrapperPtr =
    std::unique_ptr<SocketInterruptablePollWrapper>;

class MySQLTCPHandler : public Poco::Net::TCPServerConnection {
 public:
  MySQLTCPHandler(Poco::Net::StreamSocket& socket)
      : Poco::Net::TCPServerConnection(socket) {}
  using Poco::Net::TCPServerConnection::TCPServerConnection;

  void sendClientPacket(MySQLPacket* packet) {
    socket().sendBytes(packet->buff, packet->length());
  }

  void receiveClientPacket(MySQLPacket* packet) {
    UInt32 payload_len = 0;
    UInt8 seq_id = 0;
    if (socket().receiveBytes(&payload_len, 3) > 0) {
      socket().receiveBytes(&seq_id, 1);
      LOG_TRACE("Receive: payload_len:%d, seq_id:%d", payload_len, seq_id);
      packet->packUInt24(payload_len);
      packet->packUInt8(seq_id);
      socket().receiveBytes(packet->write_pos, payload_len);
    }
  }

  bool connectionPhase() {
    try {
      MySQLPacket packet;
      MySQLHandShake handshake;
      handshake.pack(&packet);
      LOG_INFO("%s", handshake.toString().c_str());
      sendClientPacket(&packet);
      packet.resetPacket();
      receiveClientPacket(&packet);
      MySQLHandShakeResponse handshake_response;
      handshake_response.unpack(&packet);
      LOG_INFO("%s", handshake_response.toString().c_str());
      packet.resetPacket();
      MySQLOKResponse ok;
      LOG_INFO("%s", ok.toString().c_str());
      ok.setSeqID(2);
      ok.pack(&packet);
      sendClientPacket(&packet);
    } catch (std::exception& e) {
      LOG_INFO("connectionPhase with exception:%s", e.what());
      return false;
    }
    return true;
  }

  MySQLCommand getRequestCommandType(MySQLPacket& handle_packet) {
    return static_cast<MySQLCommand>(
        *(handle_packet.read_pos + MYSQL_PACKET_HEAD_SIZE));
  }
  const char* getCommandSQL(MySQLPacket& handle_packet) {
    return (char*)(handle_packet.read_pos + MYSQL_PACKET_HEAD_SIZE + 1);
  }
  void handleRequest(MySQLPacket& handle_packet);
  void handleQueryCommand(MySQLPacket& handle_packet);

  void run() override {
    if (!connectionPhase()) return;

    try {
      Poco::Timespan timeout = 5;
      poll_wrapper = std::make_unique<SocketInterruptablePollWrapper>(socket());
      while (true) {
        PollResult result = poll_wrapper->poll(timeout);
        if (result.has_requests) {
          MySQLPacket packet;
          receiveClientPacket(&packet);
          handleRequest(packet);
        }
      }
    } catch (Poco::Exception& e) {
      LOG_INFO("error:%s", e.what());
    } catch (std::exception& e) {
      LOG_INFO("std::error:%s", e.what());
    }
  }

 private:
  UInt32 packet_num = 0;
  std::string _format = "%dd %H:%M:%S.%i";
  SocketInterruptablePollWrapperPtr poll_wrapper;
};

}  // namespace keeper_server
